package com.thp.bigdata.rjoin.findFriends;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.Job;

/**
 * 这个启动方法是我想使用ChainMapper 跟 ChainReducer 来设置
 * 我想的是我不运行两次，直接一次跑完 一个Map,一个Reduce,一个Map,一个Reduce
 * 但是我发现目前我使用的这种做法是行不通的。
 * 因为只能运行在一个MapReduce程序中使用一个Reducer
 * @author 汤小萌
 *
 */
public class ChainRun {
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		
		conf.set("mapreduce.framework.name", "local");
		conf.set("fs.defaultFS", "file:///");
		
		Job job = Job.getInstance(conf);
		
		job.setJar("f:/rjoin.jar");
		
		/*ChainMapper.addMapper(job, CommonFriendsStepOneMapper.class, 
				LongWritable.class, Text.class, Text.class, Text.class, conf);
		ChainReducer.setReducer(job, CommonFriendsStepOneReducer.class, 
				Text.class, Text.class, Text.class, Text.class, conf);
		
		ChainMapper.addMapper(job, CommonFriendsStepTwoMapper1.class, 
				Text.class, Text.class, Text.class, Text.class, conf);*/
		
		
		
		/*job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);*/
		
		// job.setInputFormatClass(TextInputFormat.class);
		// job.setOutputFormatClass(TextOutputFormat.class);
		
		
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		job.setMapperClass(CommonFriendsStepOneMapper.class);
		job.setReducerClass(CommonFriendsStepOneReducer.class);
		
		
		FileInputFormat.setInputPaths(job, new Path("f:/commonFriends1/input"));
		FileOutputFormat.setOutputPath(job, new Path("f:/commonFriends1/outpu_1"));
		
		job.waitForCompletion(true);
		
		////////////////////////////////////////////////////////////////////////
		Configuration conf1 = new Configuration();
		
		conf.set("mapreduce.framework.name", "local");
		conf.set("fs.defaultFS", "file:///");
		
		Job newJob = Job.getInstance(conf1);
		
		newJob.setJar("f:/rjoin.jar");
		
		newJob.setOutputKeyClass(Text.class);
		newJob.setOutputValueClass(Text.class);
		
		newJob.setMapperClass(CommonFriendsStepTwoMapper.class);
		newJob.setReducerClass(CommonFriendsStepTwoReducer.class);
		
		
		FileInputFormat.setInputPaths(job, new Path("f:/commonFriends/output_1"));
		FileOutputFormat.setOutputPath(job, new Path("f:/commonFriends/output_2"));
		
		boolean res = newJob.waitForCompletion(true);
		
		
		System.exit(res ? 0 : 1);
		
	}
	
	
}
